<?php

namespace App\Console\Commands\Canal;

use App\Kuafu\CouponExt\Model\OrderCouponExtModel;
use App\Kuafu\Order\Model\OrderCouponModel;
use App\Kuafu\Order\Model\OrderDetailModel;
use App\Kuafu\Order\Model\OrderDiscountModel;
use App\Kuafu\Order\Model\OrderModel;
use App\Kuafu\Shop\Model\ShopMonitorDataCHModel;
use App\Library\FeiShu\SendFs;
use Bavix\LaravelClickHouse\Database\Eloquent\Builder;
use Com\Alibaba\Otter\Canal\Protocol\EntryType;
use Com\Alibaba\Otter\Canal\Protocol\EventType;
use Com\Alibaba\Otter\Canal\Protocol\RowChange;
use Illuminate\Console\Command;
use Illuminate\Support\Facades\Config;
use Illuminate\Support\Facades\DB;
use Illuminate\Support\Facades\Log;
use xingwenge\canal_php\CanalClient;
use xingwenge\canal_php\CanalConnectorFactory;
use xingwenge\canal_php\Fmt;

class SyncRmsToTiger extends Command
{
    /**
     * The name and signature of the console command.
     *
     * @var string
     */
    protected $signature = 'canal:sync-rms-to-tiger {--skip=0}';

    /**
     * The console command description.
     *
     * @var string
     */
    protected $description = '同步 RMS 数据至 tiger
    --skip= : 跳过的 ACK 数量';

    /**
     * Create a new command instance.
     *
     * @return void
     */
    public function __construct()
    {
        parent::__construct();
    }

    /**
     * Execute the console command
     */
    public function handle()
    {
        if (!isTest()) {
            return;
        }

        $this->initQaTigerDb();
        try {
            self::sendFs('The Canal has started.');

            $skip   = $this->option('skip');
            $client = CanalConnectorFactory::createClient(CanalClient::TYPE_SOCKET_CLUE);

            $host = config('database.sync.canal.host');
            $client->connect($host);
            $client->checkValid();

            $client->subscribe(1001, 'rms-qa-01', 'kf_rms.*'); # 修改过滤
            while (true) {
                $message = $client->getWithoutAck();
                if ($skip > 0) {
                    $client->ack($message->getId());
                    $skip--;
                    continue;
                }

                if ($entries = $message->getEntries()) {
                    foreach ($entries as $entry) {
                        self::syncRms2Tiger($entry);
                    }
                    $client->ack($message->getId());
                }
            }
        } catch (\Exception $e) {
            self::sendFs('canal:sync-rms-to-tiger 异常, 非捕捉的异常 [已经阻塞后续同步了!]', self::formatException($e));
        }
    }

    private function initQaTigerDb()
    {
        Config::set('database.connections.tiger', [
            'driver'         => 'mysql',
            'url'            => '',
            'host'           => 'rm-bp10yna19hx5lj418.mysql.rds.aliyuncs.com',
            'database'       => 'kf_tiger',
            'username'       => 'kf_tiger',
            'password'       => 'mZO6aGmNrJzeZ4Xu',
            'port'           => '3306',
            'unix_socket'    => '',
            'charset'        => 'utf8mb4',
            'collation'      => 'utf8mb4_general_ci',
            'prefix'         => '',
            'prefix_indexes' => true,
            'strict'         => false,
            'engine'         => null,
            'options'        => extension_loaded('pdo_mysql') ? array_filter([
                \PDO::MYSQL_ATTR_SSL_CA => env('MYSQL_ATTR_SSL_CA'),
            ]) : [],
        ]);
    }

    private static function syncRms2Tiger($entry)
    {
        switch ($entry->getEntryType()) {
            case EntryType::TRANSACTIONBEGIN:
            case EntryType::TRANSACTIONEND:
                return;
                break;
        }

        $rowChange = new RowChange();
        $rowChange->mergeFromString($entry->getStoreValue());
        $evenType  = $rowChange->getEventType();
        $header    = $entry->getHeader();
        $tableName = $header->getTableName();


        // 只处理 kf_shop 表
        $limitTable = ['kf_shops'];

        if (!in_array($tableName, $limitTable)) {
            return;
        }

        // TODO 优化需要更新的表
        switch ($tableName) {
            case 'kf_shops':
                self::doSimpleSync($tableName, $rowChange, $evenType);
                break;
        }
    }

    private function doSimpleSync(string $tableName, RowChange $rowChange, int $evenType)
    {
        foreach ($rowChange->getRowDatas() as $rowData) {


            switch ($evenType) {
                // 新增
                case EventType::INSERT:
                    $affectRowData = self::fmtColumn($rowData->getAfterColumns());
                    DB::connection('tiger')->table($tableName)->insert($affectRowData);
                    break;
                // 硬删除
                case EventType::DELETE:
                    $affectRowData = self::fmtColumn($rowData->getBeforeColumns());
                    DB::connection('tiger')->table($tableName)->where('id', $affectRowData['id'])->delete();
                    break;
                // 更新
                case EventType::UPDATE:
                    $affectRowData = self::fmtColumn($rowData->getAfterColumns());
                    DB::connection('tiger')->table($tableName)->where('id', $affectRowData['id'])
                      ->update($affectRowData);
                    break;
                default:
                    break;
            }
        }
    }


    private static function fmtColumn($columns)
    {
        $arr = [];
        foreach ($columns as $column) {
            $arr[$column->getName()] = $column->getValue();
        }
        return $arr;
    }

    private function sendFs($title, $content = '')
    {
        $this->info($title);
        Log::channel('orderSyncSingle')->error('同步数据库错误:', ['title' => $title, 'content' => $content]);
        SendFs::newRobot()->setTitleContent('[' . date('Y-m-d H:i:s') . '] : ' . $title, $content)->send();
    }

    private static function formatException(\Exception $e)
    {
        return json_encode([
            'line'    => $e->getLine(),
            'message' => $e->getMessage(),
            'trace'   => $e->getTrace()
        ], JSON_UNESCAPED_UNICODE);

    }

}
